package com.shujia.flink.state;

import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class Demo4Tran {
    public static void main(String[] args) throws Exception {

        /*
         * 我们先实现第一版报警程序，对于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息。
         */

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        //解析数据
        DataStream<Tran> transDS = linesDS.map(line -> {
            String[] split = line.split(",");
            String id = split[0];
            double price = Double.parseDouble(split[1]);
            return new Tran(id, price);
        });

        //按照账户id进行分组
        KeyedStream<Tran, String> keyBydS = transDS.keyBy(Tran::getId);

        keyBydS.process(new KeyedProcessFunction<String, Tran, Object>() {

            ValueState<Boolean> flagState;

            @Override
            public void open(Configuration parameters) throws Exception {
                RuntimeContext context = getRuntimeContext();
                //保存前一条数据是否小于1的状态标记
                flagState = context.getState(new ValueStateDescriptor<>("flag", Types.BOOLEAN));
            }

            @Override
            public void processElement(Tran tran,
                                       KeyedProcessFunction<String, Tran, Object>.Context ctx,
                                       Collector<Object> out) throws Exception {
                /*
                 * 我们先实现第一版报警程序，对于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息。
                 */
                Double price = tran.getPrice();
                //获取状态中的数据
                Boolean flag = flagState.value();
                if (flag == null) {
                    flag = false;
                }
                if (flag) {
                    //出现大于 $500 的交易
                    if (price > 500) {
                        //输出一个报警信息。
                        System.out.println("输出告警信息");
                    }
                    flagState.update(false);
                }
                //出现小于 $1 美元的交易
                if (price < 1) {
                    flagState.update(true);
                }

            }
        });

        env.execute();


    }
}

@Data
@AllArgsConstructor
class Tran {
    private String id;
    private Double price;
}
